package com.atguigu.spark.util.req

import com.atguigu.spark.util.util.{DruidUtil, PropUtil}
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Duration, StreamingContext}

import java.sql.{Connection, PreparedStatement, ResultSet}
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object Spark_Black {
  def main(args: Array[String]): Unit = {
    val ssc: StreamingContext = new StreamingContext(new SparkConf().setMaster("local[*]")
      .setAppName("spark-streaming")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.testing.memory", "2147480000"), Duration.apply(3000))

    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> PropUtil("kafka.brokers"),
      ConsumerConfig.GROUP_ID_CONFIG -> PropUtil("kafka.group"),
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils
      .createDirectStream[String, String](ssc, LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Set(PropUtil("kafka.topic")), kafkaPara))
    kafkaDStream.map(x => {
      val str: String = x.value()
      val strs: Array[String] = str.split(" ")
      // timestamp area city userid adid
      val dayTs: Long = strs(0).toLong / 1000 / 60 / 60 / 24 * 24 * 60 * 60 * 1000
      ((dayTs, strs(3), strs(4)), 1)
    }).reduceByKey(_ + _).foreachRDD(
      rdd => {
        rdd.foreachPartition(
          iter => {
            /*获取连接 取出黑名单查询结果*/
            val buffer: ArrayBuffer[String] = ArrayBuffer()
            val conn: Connection = DruidUtil.getConnection
            val statement: PreparedStatement = conn.prepareStatement(
              """
                |select
                | *
                |from
                | black_list
                |""".stripMargin)
            val rs: ResultSet = statement.executeQuery()
            while (rs.next()) {
              val id: String = rs.getString(1) //获取第一个列的值 编号
              buffer.append((id))
            }
            // timestamp area city userid adid
            val array: Array[((Long, String, String), Int)] = iter.toArray
            /* 判断sum的值是否小于100并且不在黑名单种*/
            val array1: Array[((Long, String, String), Int)] = array.filter(
              x => {
                if (x._2 < 100) {
                  !buffer.contains(x._1._2)
                }
                else {
                  false
                }
              }
            )
            /* 看cnt+sum是否大于100*/
            val map: mutable.Map[(Long, String, String), Int] = mutable.Map[(Long, String, String), Int]()
            array1.foreach(
              x => {
                map.clear()
                val value: (Long, String, String) = x._1
                val ps: PreparedStatement = conn.prepareStatement(
                  """
                    |select
                    | dt,userid , adid , count
                    |from user_ad_count
                    | where
                    | dt = ?
                    | and userid = ?
                    | and adid = ?
                    |""".stripMargin)
                ps.setLong(1, value._1)
                ps.setString(2, value._2)
                ps.setString(3, value._3)
                val rs: ResultSet = ps.executeQuery()
                while (rs.next()) {
                  val ts: Long = rs.getLong(1)
                  val userId: String = rs.getString(2)
                  val adId: String = rs.getString(3)
                  val count: Int = rs.getInt(4)
                  map.update((ts, userId, adId), count)
                }
                val count: Int = map.getOrElse(x._1, 0) + x._2
                val stat: PreparedStatement = conn.prepareStatement(
                  s"""
                     | insert into
                     | user_ad_count
                     | values
                     | (?,?,?,?)
                     | ON DUPLICATE KEY UPDATE
                     | count= ?
                     |""".stripMargin)
                stat.setLong(1, value._1)
                stat.setString(2, value._2)
                stat.setString(3, value._3)
                stat.setInt(4, count)
                stat.setInt(5, count)
                stat.executeUpdate()
                if (count > 100) {
                  val stat: PreparedStatement = conn.prepareStatement(
                    s"""
                       | insert into black_list
                       | values (?)
                       |""".stripMargin)
                  stat.setString(1, value._2)
                  stat.executeUpdate()
                }
              }
            )
            conn.close()
          }
        )
      }
    )
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
